#include "all.h"

const char *filename = "/home/hq/node.json"; // 点表地址
const char *savejson = "/home/hq/save.json"; // 点表存放地址
struct shm_param para;                       // 共享内存结构体
struct CJSON jsonRejion;                     // 共享内存对应结构体
struct msgQueue QueueRejion;                 // 消息队列结构体
#define ADDRESS "tcp://localhost:1883"       // 代理地址和端口号
#define CLIENTID "ExampleClientSub"          // 客户端ID
#define FaBu "AC_pub"                        // 发布主题
#define DingYue "AC_Sub"                     // 订阅主题
#define QOS 1                                // 模式
#define TIMEOUT 10000L                       // 超时检测
MQTTClient client;
MQTTClient_deliveryToken token;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
int rc;
int ch;
struct CJSON *p = NULL;
FILE *file;
char *json_str;
struct CJSON data;
int sign = -1, type = 0;
sqlite3 *db = NULL;
char histroy[2048] = "";
cJSON *root;
char buf[3697] = "";
char buf1[3697] = "";

typedef struct
{
    char key[50];
    int value;
    time_t timestamp;
} DataPoint;

DataPoint shared_memory[100];
pthread_mutex_t shared_memory_mutex = PTHREAD_MUTEX_INITIALIZER;
int running = 1; // 进程运行标志

// 函数声明
void *reporting_thread(void *arg);
void *data_storage_thread(void *arg);
void *command_processing_thread(void *arg);
void connect_mqtt();
void creatdatabaseandtable();
void insertdatabase();
void search_by_time(struct Find_data *text);
char *GainTime(int *t);
bool userRegister(struct Find_data *text);
bool userLogon(struct Find_data *text);
void setData(int key, int val);
char *getTime(char *time, char *time_str);
volatile MQTTClient_deliveryToken deliveredtoken;
// 消息传递成功调用，订阅者不会
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}
// 消息到达调用
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    if (message == NULL || message->payload == NULL)
    {
        printf("Received null message or payload\n");
        return 0;
    }

    if (message->payloadlen >= sizeof(buf))
    {
        printf("Payload size exceeds buffer limit\n");
        return 0;
    }

    char *json_payload = (char *)malloc(message->payloadlen + 1);
    if (json_payload == NULL)
    {
        printf("Memory allocation failed for payload\n");
        return 0;
    }

    memcpy(json_payload, message->payload, message->payloadlen);
    json_payload[message->payloadlen] = '\0';

    cJSON *msg = cJSON_Parse(json_payload); // 使用 malloc 分配的内存解析 JSON
    free(json_payload);                     // 解析完成后释放 json_payload

    if (msg == NULL)
    {
        printf("Failed to parse JSON payload\n");
        return -1;
    }

    json_str = cJSON_Print(msg);

    cJSON *type_item = cJSON_GetObjectItem(msg, "type");
    if (type_item && cJSON_IsNumber(type_item))
    {
        type = type_item->valueint;
    }
    else
    {
        printf("Failed to find 'type' or it's not a number.\n");
        cJSON_Delete(msg);
        return -1;
    }

    printf("%d\n", type);

    switch (type)
    {
    case 1: // 采集进程
        if (upjson_to_struct(json_str, &data) == 0)
        {
            sign = 0;
            printf("TYPE:%d\n", data.report_type);
        }
        break;
    case 2: // 命令下发
        if (parse_downjson_to_struct(json_str, &QueueRejion) == 0)
        {
            msg_queue_send("msgqueue", &QueueRejion, sizeof(QueueRejion), 0);
        }
        break;
    case 3: // 历史记录查询
    {
        struct Find_data text;
        cJSON *item = cJSON_GetObjectItem(msg, "username");
        strcpy(text.username, item->valuestring);
        item = cJSON_GetObjectItem(msg, "max");
        getTime(text.max, item->valuestring);
        item = cJSON_GetObjectItem(msg, "min");
        getTime(text.min, item->valuestring);
        printf("%s\n%s\n", text.max, text.min);
        search_by_time(&text);
    }
    break;
    case 4: // 用户登录
    {
        printf("%d\n", type);
        struct Find_data text;
        cJSON *item = cJSON_GetObjectItem(msg, "username");
        strcpy(text.username, item->valuestring);
        item = cJSON_GetObjectItem(msg, "password");
        strcpy(text.password, item->valuestring);
        item = cJSON_GetObjectItem(msg, "key");
        printf("%d\n", item->valueint);
        if (item->valueint)
        {
            userLogon(&text);
        }
        else
        {
            userRegister(&text);
        }
    }
    break;
    case 5:
    {
        cJSON *item1 = cJSON_GetObjectItem(msg, "key");
        cJSON *item2 = cJSON_GetObjectItem(msg, "val");
        printf("%d\n%d\n", item1->valueint, item2->valueint);
        setData(item1->valueint, item2->valueint);
    }
    break;

    default:
        break;
    }

    cJSON_Delete(msg); // 解析完后删除 msg 对象
    return 1;
}

// 连接丢失调用
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
    MQTTClient_disconnect(client, 1000); // 优雅地断开连接
    MQTTClient_destroy(&client);         // 销毁客户端实例
}

// 主函数
int main()
{
    pthread_t reporting_tid, storage_tid, command_tid;
    // 连接 MQTT 服务
    connect_mqtt();
    creatdatabaseandtable();
    pthread_create(&command_tid, NULL, command_processing_thread, NULL); // 命令处理线程
    // 等待点表下发
    FILE *file;
    while (1)
    {
        file = fopen("/home/hq/node.json", "r");
        if (file == NULL)
        {
            printf("等待点表下发\n");
            sleep(2);
        }
        else
        {
            break;
        }
    }

    // 创建数据库

    // 打开共享内存
    if (shm_init(&para, "shmMery", sizeof(jsonRejion)) == -1)
    {
        ERR_MSG("开辟共享内存失败");
        pthread_exit(NULL); // 发生错误时退出线程
    }
    else
    {
        printf("开辟共享内存成功\n");
    }
    // 映射
    p = shm_getaddr(&para);
    if (p == NULL)
    {
        ERR_MSG("映射失败");
        pthread_exit(NULL); // 发生错误时退出线程
    }
    else
    {
        printf("映射成功\n");
    }

    // 下发点表
    fseek(file, 0, SEEK_END);
    long length = ftell(file);
    fseek(file, 0, SEEK_SET);
    json_str = (char *)malloc(length + 1);
    fread(json_str, 1, length, file);
    json_str[length] = '\0';
    // 解析 JSON
    if (parse_upjson_to_struct(json_str, &data) == 0)
    {
        *p = data;
    }
    sign = 0;

    // 创建线程
    pthread_create(&reporting_tid, NULL, reporting_thread, NULL);  // 上报线程
    pthread_create(&storage_tid, NULL, data_storage_thread, NULL); // 数据存储线程

    // 等待线程结束
    pthread_join(reporting_tid, NULL);
    pthread_join(storage_tid, NULL);
    pthread_join(command_tid, NULL);

    return 0;
}

// 连接 MQTT
void connect_mqtt()
{
    // 创建客户端实例
    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    // 回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    // 连接代理
    MQTTClient_connect(client, &conn_opts);
    // 订阅主题
    MQTTClient_subscribe(client, FaBu, QOS);
}

// 上报线程
void *reporting_thread(void *arg)
{
    int num = 0;
    while (1)
    {
        json_str = serialize_all(&p->stm32, &p->modbus);
        if (sign == 1)
        {
            strcpy(buf, json_str);
            cJSON_ReplaceItemInObject(root, "root", cJSON_CreateString(buf));
            pubmsg.payload = buf;
            pubmsg.payloadlen = strlen(buf);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;

            pthread_mutex_lock(&lock);
            MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
            pthread_mutex_unlock(&lock);

            sign = 4;
        }
        else if (sign == 2)
        {
            if (strcmp(json_str, buf))
            {
                num++;
            }
            if (num == 2)
            {
                strcpy(buf, json_str);
                cJSON_ReplaceItemInObject(root, "root", cJSON_CreateString(buf));
                pubmsg.payload = buf;
                pubmsg.payloadlen = strlen(buf);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                pthread_mutex_lock(&lock);
                MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
                pthread_mutex_unlock(&lock);
            }
        }
        else if (sign == 3)
        {
            memset(buf, 0, 2048);
            strcpy(buf, json_str);
            cJSON_ReplaceItemInObject(root, "root", cJSON_CreateString(buf));
            pubmsg.payload = buf;
            pubmsg.payloadlen = strlen(buf);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            pthread_mutex_lock(&lock);
            MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
            pthread_mutex_unlock(&lock);
            sleep(data.report_period);
        }
        sleep(1);
    }
    return NULL;
}

// 数据存储线程
void *data_storage_thread(void *arg)
{

    while (1)
    {
        // insertdatabase(db);
        sleep(1); // 数据存储周期}
    }
    return NULL;
}
// 指令处理线程
void *command_processing_thread(void *arg)
{

    while (1)
    {
        if (sign == 0)
        {
            switch (data.report_type)
            {
            case 0:
                sign = 1;
                break;
            case 1:
                sign = 2;
                break;
            case 2:
                sign = 3;
                break;

            default:
                break;
            }
        }
        sleep(1);
    }
    return NULL;
}

// 打开或者创建数据库和打开或创建一个表
void creatdatabaseandtable()
{
    // 1.打开或者创建数据库
    db = NULL;
    if (sqlite3_open("./data.db", &db) != 0)
    {
        printf("open err:%s\n", sqlite3_errmsg(db));
        return;
    }
    printf("data.db open ok\n");
    // 2.创建两个表
    char *errmsg = NULL;
    if (sqlite3_exec(db, "create table data(laboratory_status TEXT,data_Vol float,mode int,data_TemController int,data_Tmp float,Tmp_Low int,Tmp_Hign int,data_Dehunidifier int,data_Hum float,Hum int,result TEXT,timestr TEXT,time int);", NULL, NULL, &errmsg) != 0)
    {
        printf("create err:%s\n", errmsg);
    }
    printf("data create ok\n");
    errmsg = NULL;
    if (sqlite3_exec(db, "create table user(username TEXT primary key,password TEXT);", NULL, NULL, &errmsg) != 0)
    {
        printf("create err:%s\n", errmsg);
    }
    printf("user create ok\n");
}

// 获取当前时间
char *GainTime(int *t)
{
    time_t nowtime;
    struct tm *timeinfo;
    time(&nowtime);
    *t = nowtime;
    timeinfo = localtime(&nowtime);
    static char timestr[100];
    strftime(timestr, sizeof(timestr), "%Y-%m-%d %I:%M:%S", timeinfo);
    return timestr;
}

// 记录数据和时间
void insertdatabase()
{
    char sql[2048];
    int time = 0;
    char *timestr = GainTime(&time);
    char *errmsg = NULL;
    char status[64] = "";
    if (p->modbus.data[0].val.data_switch == 0 && p->modbus.data[1].val.data_switch == 0 && p->modbus.data[2].val.data_switch == 0)
    {
        strcpy(status, "safe");
    }
    else
    {
        strcpy(status, "danger");
    }
    char result[64] = "uncertain";
    sprintf(sql, "insert into data values(\"%s\",%.2f,%d,%d,%.2f,%d,%d,%d,%.2f,%d,\"%s\",\"%s\",%d);", status, p->stm32.data[4].val.data_val_f, p->stm32.data[3].val.data_val_i, p->modbus.data[5].val.data_val_i, p->stm32.data[5].val.data_val_f, p->stm32.data[1].val.data_val_i, p->stm32.data[0].val.data_val_i, p->modbus.data[4].val.data_val_i, p->stm32.data[6].val.data_val_f, p->stm32.data[2].val.data_val_i, result, timestr, time);
    if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
    {
        printf("insert err:%s\n", errmsg);
        return;
    }
    printf("insert ok\n%s\n", sql);
}

// 用户注册
bool userRegister(struct Find_data *text)
{
    char sql[2048];
    char *errmsg = NULL;
    sprintf(sql, "insert into user values(\"%s\",\"%s\");", text->username, text->password);
    printf("sql:::::%s\n", sql);
    if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
    {
        printf("user insert err:%s\n", errmsg);
        char *his = user_up(2); // 注册失败
        printf("buf1 : %s\n", his);
        strcpy(buf1, his);
        pubmsg.payload = buf1;
        pubmsg.payloadlen = strlen(buf1);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        pthread_mutex_lock(&lock);
        MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
        pthread_mutex_unlock(&lock);
        return 0;
    }
    char *his = user_up(1); // 注册成功
    strcpy(buf1, his);
    printf("buf1 : %s\n", buf1);
    pubmsg.payload = buf1;
    pubmsg.payloadlen = strlen(buf1);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    pthread_mutex_lock(&lock);
    MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
    pthread_mutex_unlock(&lock);
    return 1;
}

// 用户登录
bool userLogon(struct Find_data *text)
{
    memset(histroy, 0, 2048);
    char s1[2048] = {};
    char **Result;
    int hang, num = 0;
    int lie;
    char sql[2048] = {};
    char *errmsg = NULL;
    sprintf(sql, "select * from user where username=\"%s\" and password=\"%s\";", text->username, text->password);
    if (sqlite3_get_table(db, sql, &Result, &hang, &lie, &errmsg) != 0)
    {
        printf("username select err:%s\n", errmsg);
    }
    if (hang > 0)
    {
        char *his = user_up(3); // 登录成功
        strcpy(buf1, his);
        pubmsg.payload = buf1;
        pubmsg.payloadlen = strlen(buf1);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        pthread_mutex_lock(&lock);
        MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
        pthread_mutex_unlock(&lock);
        return 1;
    }
    char *his = user_up(4); // 登录失败
    strcpy(buf1, his);
    pubmsg.payload = buf1;
    pubmsg.payloadlen = strlen(buf1);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    pthread_mutex_lock(&lock);
    MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
    pthread_mutex_unlock(&lock);
    return 0;
}

// 查历史记录
void search_by_time(struct Find_data *text)
{
    memset(histroy, 0, 2048);
    char s1[2048] = {};
    char **Result;
    int hang, num = 0;
    int lie;
    char sql[2048] = {};
    char *errmsg = NULL;
    struct history_up his_up;
    his_up.type = 114514;
    // 按时间查历史记录
    sprintf(sql, "select * from data where time >= \"%s\" and time <=\"%s\";", text->min, text->max);
    if (sqlite3_get_table(db, sql, &Result, &hang, &lie, &errmsg) != 0)
    {
        printf("time select err:%s\n", errmsg);
        return;
    }
    if (hang > 0) // 查到信息
    {
        // 打印标签
        // sprintf(histroy, "查询到的历史记录：\n%s  %s  %s  %-14s  %-9s  %s  %s  %s  %s  %s  %s  %s  %s", Result[0], Result[1], Result[2], Result[3], Result[4], Result[5], Result[6], Result[7], Result[8], Result[9], Result[10], Result[11], Result[12]);
        printf("hang:%d\n", hang);
        while ((hang - num) > 0)
        {
            printf("hang -num:%d\n", hang - num);
            printf("num:%d\n", num);
            num++;
            // sprintf(s1, "\n%-9s %-18s %-9s %-5s %-15s %-10s %s %s %s %s %s %s %s %s", Result[lie * num], Result[lie * num + 1], Result[lie * num + 2], Result[lie * num + 3], Result[lie * num + 4], Result[lie * num + 5], Result[lie * num + 6], Result[lie * num + 7], Result[lie * num + 8], Result[lie * num + 9], Result[lie * num + 10], Result[lie * num + 11], Result[lie * num + 12]);
            // // 将数据接到标签后面并换行
            // strcat(histroy, s1);
            strcpy(his_up.mode, Result[lie * num + 11]);
            strcpy(his_up.data_Vol, Result[lie * num + 1]);
            strcpy(his_up.data_Tmp, Result[lie * num + 4]);
            strcpy(his_up.data_Hum, Result[lie * num + 8]);
            printf("%s\n%s\n%s\n%s\n", his_up.mode, his_up.data_Vol, his_up.data_Tmp, his_up.data_Hum);
            char *his = serialize_up(&his_up);
            strcpy(buf1, his);
            pubmsg.payload = buf1;
            pubmsg.payloadlen = strlen(buf1);
            pubmsg.qos = 0;
            pubmsg.retained = 0;
            printf("%s\n", buf1);
            pthread_mutex_lock(&lock);
            MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
            pthread_mutex_unlock(&lock);
            printf("---%d\n", __LINE__);
            free(his);
            // sleep(1);
        }
        // MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
    }
    // else // 未查到信息，可能是按用户名查历史记录
    // {
    //     sprintf(sql, "select * from data where username=\"%s\";", text->username);
    //     if (sqlite3_get_table(db, sql, &Result, &hang, &lie, &errmsg) != 0)
    //     {
    //         printf("username select err:%s\n", errmsg);
    //         return;
    //     }
    //     if (hang > 0)
    //     {
    //         // sprintf(histroy, "查询到的历史记录：\n%s  %s  %s  %s  %-14s  %-9s  %s  %s  %s  %s  %s  %s  %s  %s", Result[0], Result[1], Result[2], Result[3], Result[4], Result[5], Result[6], Result[7], Result[8], Result[9], Result[10], Result[11], Result[12], Result[13]);
    //     while ((hang - num) > 0)
    //     {
    //         num++;
    //         // sprintf(s1, "\n%-9s %-18s %-9s %-5s %-15s %-10s %s %s %s %s %s %s %s %s", Result[lie * num], Result[lie * num + 1], Result[lie * num + 2], Result[lie * num + 3], Result[lie * num + 4], Result[lie * num + 5], Result[lie * num + 6], Result[lie * num + 7], Result[lie * num + 8], Result[lie * num + 9], Result[lie * num + 10], Result[lie * num + 11], Result[lie * num + 12], Result[lie * num + 13]);
    //         // // 将数据接到标签后面并换行
    //         // strcat(histroy, s1);
    //         strcpy(his_up.mode,Result[lie * num + 3]);
    //         strcpy(his_up.mode,Result[lie * num + 2]);
    //         strcpy(his_up.mode,Result[lie * num + 5]);
    //         strcpy(his_up.mode,Result[lie * num + 9]);
    //         char *his =serialize_up(&his_up);
    //         strcpy(buf1, his);
    //         cJSON_ReplaceItemInObject(root, "root", cJSON_CreateString(buf1));
    //         pubmsg.payload = buf1;
    //         pubmsg.payloadlen = strlen(buf1);
    //         pubmsg.qos = QOS;
    //         pubmsg.retained = 0;
    //         MQTTClient_publishMessage(client, DingYue, &pubmsg, &token);
    //     }
    //     }
    else
    {
        strcpy(histroy, "未查到历史记录！！！");
    }
    // }
    printf("%s\n", histroy);
    return;
}

// 设置数据
void setData(int key, int val)
{
    struct msgQueue msgtext;
    msgtext.text.state_Humidifier = p->modbus.data[0].val.data_val_i;
    msgtext.text.state_Dehumidifier = p->modbus.data[1].val.data_val_i;
    msgtext.text.state_TemController = p->modbus.data[2].val.data_val_i;
    msgtext.text.data_Humidifier = p->modbus.data[3].val.data_val_i;
    msgtext.text.data_Dehumidifier = p->modbus.data[4].val.data_val_i;
    msgtext.text.data_TemController = p->modbus.data[5].val.data_val_i;
    msgtext.text.Tmp_Hign = p->stm32.data[0].val.data_val_i;
    msgtext.text.Tmp_Low = p->stm32.data[1].val.data_val_i;
    msgtext.text.Hum = p->stm32.data[2].val.data_val_i;
    msgtext.text.mode = p->stm32.data[3].val.data_val_i;
    msgtext.text.mode_aircondition = p->stm32.data[8].val.data_val_i;
    msgtext.key = key;
    switch (msgtext.key)
    {
    case 101:
        msgtext.text.state_Humidifier = val;
        break;
    case 102:
        msgtext.text.state_Dehumidifier = val;
        break;
    case 103:
        msgtext.text.state_TemController = val;
        break;
    case 104:
        msgtext.text.data_Humidifier = val;
        break;
    case 105:
        msgtext.text.data_Dehumidifier = val;
        break;
    case 106:
        msgtext.text.data_TemController = val;
        break;
    case 301:
        msgtext.text.Tmp_Hign = val;
        break;
    case 302:
        msgtext.text.Tmp_Low = val;
        break;
    case 303:
        msgtext.text.Hum = val;
        break;
    case 304:
        msgtext.text.mode = val;
        break;
    case 309:
        msgtext.text.mode_aircondition = val;
        break;

    default:
        break;
    }
    if (msgtext.key / 100 == 1)
    {
        // printf("asdf::::::::::::::::::::::::::%d\n", p->stm32.data[2].val.data_val_i);
        msgtext.key = 100;
    }
    else
    {
        // printf("asdf::::::::::::::::::::::::::%d\n", p->stm32.data[2].val.data_val_i);
        msgtext.key = 200;
    }
    // printf("key:::::::::::::::::::::::%ld  val:%d\n", msgtext.key, msgtext.text.Tmp_Hign);
    // printf("key:::::::::::::::::::::::::%ld  val:%d\n", msgtext.key, msgtext.text.Tmp_Low);
    // printf("key:::::::::::::::::::::::::%ld  val:%d\n", msgtext.key, msgtext.text.Hum);
    // printf("msgtextsize::::::::::::::::::%ld\n", sizeof(msgtext));
    msg_queue_send("msgqueue", &msgtext, sizeof(msgtext), 0);
}

// 转换位秒数
char *getTime(char *time, char *time_str)
{
    struct tm tm_time = {0};
    time_t timestamp;

    // 将字符串解析为 tm 结构
    if (strptime(time_str, "%Y/%m/%d %H:%M", &tm_time) == NULL)
    {
        fprintf(stderr, "时间格式错误\n");
        return NULL;
    }

    // 转换为时间戳（秒数）
    timestamp = mktime(&tm_time);
    if (timestamp == -1)
    {
        fprintf(stderr, "时间转换失败\n");
        return NULL;
    }
    sprintf(time, "%d", timestamp);
    return time;
}
